package flink.connector.alarm;

import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.FactoryUtil;

import org.junit.Assert;

import java.util.HashSet;
import java.util.Set;

import static flink.connector.alarm.WechatAlarmConnectorOptions.*;

/**
 * {@link DynamicTableSinkFactory} for {@link WechatAlarmDynamicTableSink} 用于发送到企业微信报警的 connector
 * 所有发送报警的都可以共用这个 connector 使用:
 *
 * <p>create table wechat_alarm ( level string, type string, details string, alarmTime string,
 * job_name string, job_id string, metricName string ) WITH ( 'connector' = 'wechat-alarm', 'url' =
 * 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=69393f4a-f847-4c80-ad13-a4c0f86d49af',
 * 'msgtype' = 'MARKDOWN' );
 *
 * <p>insert into wechat_alarm select * from table 目前支持两种类型的 msgtype: MARKDOWN/TEXT 如果不设置默认为
 * MARKDOWN
 *
 * @author jasonlee
 */
@Internal
public class WechatAlarmTableSinkFactory implements DynamicTableSinkFactory {
    public static final String IDENTIFIER = "wechat-alarm";

    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> requiredOptions = new HashSet<>();
        requiredOptions.add(URL);
        return requiredOptions;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> optionSet = new HashSet<>();
        // MSGTYPE 可以不配置,默认使用 Markdown 格式发送报警
        optionSet.add(MSGTYPE);
        optionSet.add(MENTIONED_LIST);
        return optionSet;
    }

    @Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        helper.validate();
        ReadableConfig options = helper.getOptions();
        Assert.assertNotNull("URL 不能为空,必须配置 URL 地址", options.getOptional(URL).get());
        return new WechatAlarmDynamicTableSink(options);
    }
}
